package com.shujia.mr.kqzl2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;

public class KQZLReducer2 extends Reducer<Text, Text, Text, LongWritable> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //<"1001",["#20180216-91","$北京-万寿西宫","#20180217-86",...]>
        /*
            最终效果：
            20180216 北京 万寿西宫 91
            20180217 北京 万寿西宫 86
            ..
         */
        String city = null;
        //定义一个集合保存该监测点每一天的平均值以及日期信息
        ArrayList<String> list = new ArrayList<>();

        //处理迭代器的信息，得到该监测点的城市信息以及每一天的日期平均值信息。
        for (Text value : values) {
            String info = value.toString();
            if (info.startsWith("$")) {
                city = info.substring(1).replace("-", " "); //北京 万寿西宫
            } else if (info.startsWith("#")) {
                list.add(info.substring(1));
            }
        }

        //关联数据
        for (String s : list) {
            String[] strings = s.split("-");
            String date = strings[0];
            long pm25 = Long.parseLong(strings[1]);
            context.write(new Text(date + " " + city), new LongWritable(pm25));
        }

    }
}
